#ifndef MONITOR_NEW_CORE_MONITOR_QUEUE_H_
#define MONITOR_NEW_CORE_MONITOR_QUEUE_H_

#include <atomic>
#include <condition_variable>
#include <deque>
#include <memory>
#include <mutex>
#include <optional>
#include <vector>

#include "build/proto_src/system_info.pb.h"

namespace sysinfo {
namespace monitor {

/******************************************
 *              基类模板                   *
 ******************************************/
template <typename ValueType, typename QueueType>
class MonitorQueueBase {
 public:
  MonitorQueueBase()
      : size_(0), finish_(false), queue_(std::make_unique<QueueType>()) {}
  explicit MonitorQueueBase(size_t size)
      : size_(size),
        finish_(false),
        queue_(std::make_unique<QueueType>(size)) {}

  size_t Capacity() const { return size_; }

  size_t Size() const { return queue_->size(); }

  bool Empty() const { return queue_->empty(); }

  bool IsFinish() const { return finish_.load(std::memory_order_relaxed); }

  void NotifyAll() { queue_->notify_all(); }

  ValueType Take() { return queue_->take(); }

  std::optional<ValueType> Take(int64_t timeout_ms) {
    return queue_->take(timeout_ms);
  }

  std::vector<ValueType> TakeBatch(size_t batch_size) {
    return queue_->takeBatch(batch_size);
  }

  void Put(const ValueType& item) {
    queue_->put(item);
    cv_.notify_one();
  }

  void Put(ValueType&& item) {
    queue_->put(std::forward<ValueType>(item));
    cv_.notify_one();
  }

  void SetFinish(bool value) {
    finish_.store(value, std::memory_order_relaxed);
    cv_.notify_all();
  }

  void Wait() { Wait(size_ == 0 ? 1 : size_ - (size_ >> 2)); }

  void Wait(size_t num) {
    std::unique_lock<std::mutex> lock(tmux_);
    cv_.wait(lock, [this, num] { return queue_->size() >= num || finish_; });
  }

 private:
  size_t size_;

  std::atomic<bool> finish_;
  std::unique_ptr<QueueType> queue_;

  std::mutex tmux_;
  std::condition_variable cv_;
};

/******************************************
 *           BlockingQueue实现             *
 ******************************************/
template <typename ValueType>
class BlockingQueue {
 public:
  using MutexUniqueLock = std::unique_lock<std::mutex>;
  static constexpr size_t kMaxQueueSize = 1000;

  explicit BlockingQueue() : mutex_(), not_empty_(), not_full_(), is_finish_(false) {}

  BlockingQueue(const BlockingQueue&) = delete;
  BlockingQueue& operator=(const BlockingQueue&) = delete;

  void put(const ValueType& item) {
    {
      MutexUniqueLock lock(mutex_);
      not_full_.wait(lock, [this]() { return queue_.size() < kMaxQueueSize || is_finish_; });
      if (!is_finish_) {
        queue_.emplace_back(item);
      }
    }
    not_empty_.notify_one();
  }

  void put(ValueType&& item) {
    {
      MutexUniqueLock lock(mutex_);
      not_full_.wait(lock, [this]() { return queue_.size() < kMaxQueueSize || is_finish_; });
      if (!is_finish_) {
        queue_.emplace_back(std::forward<ValueType>(item));
      }
    }
    not_empty_.notify_one();
  }

  ValueType take() {
    ValueType item;
    {
      MutexUniqueLock lock(mutex_);
      not_empty_.wait(lock, [this]() { return !queue_.empty() || is_finish_; });
      if (!queue_.empty() && !is_finish_) {
        item = std::move(queue_.front());
        queue_.pop_front();
      }
    }
    not_full_.notify_one();
    return item;
  }

  std::optional<ValueType> take(int64_t timeout_ms) {
    std::optional<ValueType> item;
    {
      MutexUniqueLock lock(mutex_);
      if (!not_empty_.wait_for(lock, std::chrono::milliseconds(timeout_ms),
                               [this]() { return !queue_.empty() || is_finish_; })) {
        return std::nullopt;
      }
      if (!queue_.empty() && !is_finish_) {
        item = std::move(queue_.front());
        queue_.pop_front();
      }
    }
    not_full_.notify_one();
    return item;
  }

  std::vector<ValueType> takeBatch(size_t batch_size) {
    std::vector<ValueType> vecs;
    {
      MutexUniqueLock lock(mutex_);
      not_empty_.wait(
          lock, [this, batch_size]() { return queue_.size() >= batch_size || is_finish_; });
      while (batch_size-- > 0 && !queue_.empty() && !is_finish_) {
        vecs.emplace_back(std::move(queue_.front()));
        queue_.pop_front();
      }
    }
    not_full_.notify_one();
    return vecs;
  }

  bool empty() const {
    MutexUniqueLock lock(mutex_);
    return queue_.empty();
  }

  size_t size() const {
    MutexUniqueLock lock(mutex_);
    return queue_.size();
  }

  void notify_all() {
    not_empty_.notify_all();
    not_full_.notify_all();
  }

  void SetFinish(bool finish) {
    {
      MutexUniqueLock lock(mutex_);
      is_finish_ = finish;
    }
    notify_all();
  }

  bool IsFinish() const {
    MutexUniqueLock lock(mutex_);
    return is_finish_;
  }

 private:
  mutable std::mutex mutex_;
  std::condition_variable not_empty_;
  std::condition_variable not_full_;
  std::deque<ValueType> queue_;
  bool is_finish_;
};

/******************************************
 *        BoundedBlockingQueue实现         *
 ******************************************/
template <typename ValueType>
class BoundedBlockingQueue {
 public:
  using MutexUniqueLock = std::unique_lock<std::mutex>;

  explicit BoundedBlockingQueue(size_t capacity)
      : mutex_(), not_empty_(), not_full_(), capacity_(capacity), is_finish_(false) {}

  BoundedBlockingQueue(const BoundedBlockingQueue&) = delete;
  BoundedBlockingQueue& operator=(const BoundedBlockingQueue&) = delete;

  void put(const ValueType& item) {
    {
      MutexUniqueLock lock(mutex_);
      not_full_.wait(lock, [this]() { return queue_.size() < capacity_ || is_finish_; });
      if (!is_finish_) {
        queue_.emplace_back(item);
      }
    }
    not_empty_.notify_one();
  }

  void put(ValueType&& item) {
    {
      MutexUniqueLock lock(mutex_);
      not_full_.wait(lock, [this]() { return queue_.size() < capacity_ || is_finish_; });
      if (!is_finish_) {
        queue_.emplace_back(std::forward<ValueType>(item));
      }
    }
    not_empty_.notify_one();
  }

  ValueType take() {
    ValueType item;
    {
      MutexUniqueLock lock(mutex_);
      not_empty_.wait(lock, [this]() { return !queue_.empty() || is_finish_; });
      if (!queue_.empty() && !is_finish_) {
        item = std::move(queue_.front());
        queue_.pop_front();
      }
    }
    not_full_.notify_one();
    return item;
  }

  std::optional<ValueType> take(int64_t timeout_ms) {
    std::optional<ValueType> item;
    {
      MutexUniqueLock lock(mutex_);
      if (!not_empty_.wait_for(lock, std::chrono::milliseconds(timeout_ms),
                               [this]() { return !queue_.empty() || is_finish_; })) {
        return std::nullopt;
      }
      if (!queue_.empty() && !is_finish_) {
        item = std::move(queue_.front());
        queue_.pop_front();
      }
    }
    not_full_.notify_one();
    return item;
  }

  std::vector<ValueType> takeBatch(size_t batch_size) {
    std::vector<ValueType> vecs;
    {
      MutexUniqueLock lock(mutex_);
      not_empty_.wait(
          lock, [this, batch_size]() { return queue_.size() >= batch_size || is_finish_; });
      while (batch_size-- > 0 && !queue_.empty() && !is_finish_) {
        vecs.emplace_back(std::move(queue_.front()));
        queue_.pop_front();
      }
    }
    not_full_.notify_one();
    return vecs;
  }

  bool empty() const {
    MutexUniqueLock lock(mutex_);
    return queue_.empty();
  }

  bool full() const {
    MutexUniqueLock lock(mutex_);
    return queue_.size() >= capacity_;
  }

  size_t size() const {
    MutexUniqueLock lock(mutex_);
    return queue_.size();
  }

  size_t capacity() const {
    MutexUniqueLock lock(mutex_);
    return capacity_;
  }

  void notify_all() {
    not_empty_.notify_all();
    not_full_.notify_all();
  }

  void SetFinish(bool finish) {
    {
      MutexUniqueLock lock(mutex_);
      is_finish_ = finish;
    }
    notify_all();
  }

  bool IsFinish() const {
    MutexUniqueLock lock(mutex_);
    return is_finish_;
  }

 private:
  mutable std::mutex mutex_;
  std::condition_variable not_empty_;
  std::condition_variable not_full_;
  std::deque<ValueType> queue_;
  size_t capacity_;
  bool is_finish_;
};

/******************************************
 *        CircularBlockingQueue实现        *
 ******************************************/
template <typename ValueType>
class CircularBlockingQueue {
 public:
  using MutexUniqueLock = std::unique_lock<std::mutex>;

  explicit CircularBlockingQueue(size_t capacity)
      : mutex_(), not_empty_(), capacity_(capacity), queue_(), is_finish_(false) {}

  CircularBlockingQueue(const CircularBlockingQueue&) = delete;
  CircularBlockingQueue& operator=(const CircularBlockingQueue&) = delete;

  void put(const ValueType& item) {
    {
      MutexUniqueLock lock(mutex_);
      if (queue_.size() >= capacity_) {
        // 如果队列已满，移除最旧的元素
        queue_.pop_front();
      }
      queue_.push_back(item);
    }
    not_empty_.notify_one();
  }

  void put(ValueType&& item) {
    {
      MutexUniqueLock lock(mutex_);
      if (queue_.size() >= capacity_) {
        // 如果队列已满，移除最旧的元素
        queue_.pop_front();
      }
      queue_.push_back(std::forward<ValueType>(item));
    }
    not_empty_.notify_one();
  }

  ValueType take() {
    ValueType item;
    {
      MutexUniqueLock lock(mutex_);
      not_empty_.wait(lock, [this]() { return !queue_.empty() || is_finish_; });
      if (!queue_.empty() && !is_finish_) {
        item = std::move(queue_.front());
        queue_.pop_front();
      }
    }
    return item;
  }

  std::optional<ValueType> take(int64_t timeout_ms) {
    std::optional<ValueType> item;
    {
      MutexUniqueLock lock(mutex_);
      if (!not_empty_.wait_for(lock, std::chrono::milliseconds(timeout_ms),
                               [this]() { return !queue_.empty() || is_finish_; })) {
        return std::nullopt;
      }
      if (!queue_.empty() && !is_finish_) {
        item = std::move(queue_.front());
        queue_.pop_front();
      }
    }
    return item;
  }

  std::vector<ValueType> takeBatch(size_t batch_size) {
    std::vector<ValueType> vecs;
    {
      MutexUniqueLock lock(mutex_);
      not_empty_.wait(
          lock, [this, batch_size]() { return queue_.size() >= batch_size || !queue_.empty() || is_finish_; });
      size_t actual_batch_size = std::min(batch_size, queue_.size());
      for (size_t i = 0; i < actual_batch_size; ++i) {
        if (!queue_.empty() && !is_finish_) {
          vecs.emplace_back(std::move(queue_.front()));
          queue_.pop_front();
        }
      }
    }
    return vecs;
  }

  bool empty() const {
    MutexUniqueLock lock(mutex_);
    return queue_.empty();
  }

  size_t size() const {
    MutexUniqueLock lock(mutex_);
    return queue_.size();
  }

  size_t capacity() const {
    MutexUniqueLock lock(mutex_);
    return capacity_;
  }

  void notify_all() { 
    not_empty_.notify_all(); 
  }

  void SetFinish(bool finish) {
    {
      MutexUniqueLock lock(mutex_);
      is_finish_ = finish;
    }
    notify_all();
  }

  bool IsFinish() const {
    MutexUniqueLock lock(mutex_);
    return is_finish_;
  }

  void flush() {
    MutexUniqueLock lock(mutex_);
    queue_.clear();
  }

 private:
  mutable std::mutex mutex_;
  std::condition_variable not_empty_;
  size_t capacity_;
  std::deque<ValueType> queue_;
  bool is_finish_;
};

/******************************************
 *             无界队列特化                *
 ******************************************/
template <typename ValueType>
class MonitorQueue
    : public MonitorQueueBase<ValueType, BlockingQueue<ValueType>> {
 public:
  // 默认构造无界队列（无需容量参数）
  MonitorQueue() : MonitorQueueBase<ValueType, BlockingQueue<ValueType>>() {}
};

/******************************************
 *             有界队列特化                *
 ******************************************/
template <typename ValueType>
class BoundedMonitorQueue
    : public MonitorQueueBase<ValueType, BoundedBlockingQueue<ValueType>> {
 public:
  // 强制必须提供容量参数
  explicit BoundedMonitorQueue(size_t size)
      : MonitorQueueBase<ValueType, BoundedBlockingQueue<ValueType>>(size) {}

  // 禁用默认构造
  BoundedMonitorQueue() = delete;
};

/******************************************
 *             有界 LRU 队列特化           *
 ******************************************/
template <typename ValueType>
class CircularMonitorQueue
    : public MonitorQueueBase<ValueType, CircularBlockingQueue<ValueType>> {
 public:
  // 强制必须提供容量参数
  explicit CircularMonitorQueue(size_t size)
      : MonitorQueueBase<ValueType, CircularBlockingQueue<ValueType>>(size) {}

  // 禁用默认构造
  CircularMonitorQueue() = delete;
};

// 使用模板别名简化类型定义
template<typename T>
using SystemInfoQueue = std::shared_ptr<BoundedMonitorQueue<T>>;
template<typename T>
using SystemInfoCircularQueue = std::shared_ptr<CircularMonitorQueue<T>>;

}  // namespace monitor
}  // namespace qcraft

#endif  // MONITOR_NEW_CORE_MONITOR_QUEUE_H_